package org.korosoft.rusalad.syncnrun.messaging;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Syncnrun message router.
 *
 * @author Dmitry Korotkov
 * @since 1.0
 */
public class Messenger {
    public static final Log log = LogFactory.getLog(Messenger.class);
    public static final int MAX_QUEUE_AGE = 60000;

    private final ConcurrentMap<UUID, MessageRoute> routes;
    private final MessageProcessor sender;

    /**
     * Creates a new Messenger instance.
     *
     * @param sender Message sender for outgoing messages.
     */
    public Messenger(MessageProcessor sender) {
        routes = new ConcurrentHashMap<UUID, MessageRoute>();
        this.sender = sender;
    }

    public void sendEndOfTransfer() {
        for (MessageRoute route : routes.values()) {
            MessageProcessor hotReceiver = route.getHotReceiver();
            if (hotReceiver instanceof StateAwareMessageProcessor) {
                try {
                    ((StateAwareMessageProcessor) hotReceiver).transferEnded(route);
                } catch (RuntimeException e) {
                    log.error(e.getMessage(), e);
                }
            }
        }
    }

    /**
     * Atomically creates or returns existing {@link MessageRoute} for route "name".
     * A sender is specified in {@link Messenger#Messenger(MessageProcessor)} is used as route sender.
     *
     * @param name Route name.
     * @return Message Route.
     */
    public MessageRoute getRoute(UUID name) {
        MessageRoute route = routes.get(name);
        if (route == null) {
            route = new MessageRoute(name, sender);
            MessageRoute olderRoute = routes.putIfAbsent(name, route);
            // If we lost race condition, just forget created route
            if (olderRoute != null) {
                route = olderRoute;
            }
        }
        return route;
    }
}
